package org.weishe.grandson;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

public class GrandsonReducer extends Reducer<Text, Text, Text, Text> {

	private Text k = new Text();
	private Text v = new Text();

	@Override
	protected void reduce(Text key, Iterable<Text> values, Reducer<Text, Text, Text, Text>.Context context)
			throws IOException, InterruptedException {

		List<String> l1 = new ArrayList<String>();
		List<String> l2 = new ArrayList<String>();

		for (Text t : values) {
			String str[] = t.toString().split("_");

			if (str[1].equals("1")) {
				l1.add(str[0]);
			}

			if (str[1].equals("2")) {
				l2.add(str[0]);
			}

			// context.write(key,t);
		}
		for (String s : l2) {
			String g = "";
			for (String gs : l1) {
				g = g + "," + gs;
			}
			k.set(s);

			if (g.length() > 2) {
				v.set(g.subSequence(1, g.length()).toString());
				context.write(k, v);
			}
		}

	}

}
